#!/bin/bash
#设置任务异常退出
set -eo pipefail
dt='{{ execution_date | cst_ds }}'
name='网点退件率汇总'
# 在退出时清理文件
trap "rm -f /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv && echo 'removed tmp files'" EXIT
# 创建txt文件
touch /tmp/${name}_${dt}.txt
#将中文字段名输入文件顶部
#将查询结果追加到文件中
{
  echo -e "扫描日期\t扫描网点\t扫描运单量\t退件量\t退件率\t">/tmp/${name}_${dt}.txt;
  spark-sql --driver-memory 8G --num-executors 10 --executor-memory 12g --executor-cores 6  --conf spark.executor.memoryOverhead=4g --conf spark.driver.maxResultSize=3g -S -e "
with tmp_scan_jisan as (
select
 max(network_type) as network_type
,waybill_no
,scan_site_code
,max(scan_site) as scan_site
,max(dt) as dt
,max(scan_time) as scan_time
from (
  select
       '集散点' as network_type
      ,'arrive' as scan_type
      ,waybill_no
      ,scan_site
      ,scan_site_code
      ,scan_time
      ,dt
  from jms_dwd.dwd_tab_barscan_sitearrival_base_dt --集散到件
  where dt >= date_add('{{ execution_date | cst_ds }}',-7) and dt <= '{{ execution_date | cst_ds }}'
  and network_type = 3    --集散点
  and waybill_no is not null
union all
  select
       '集散点' as network_type
      ,'send'   as scan_type
      ,waybill_no
      ,scan_site
      ,scan_site_code
      ,scan_time
      ,dt
  from jms_dwd.dwd_tab_barscan_sitesend_base_dt  --集散发件
  where dt >= date_add('{{ execution_date | cst_ds }}',-7) and dt <= '{{ execution_date | cst_ds }}'
  and network_type = 3    --集散点
  and waybill_no is not null
) tmp group by waybill_no,scan_site_code
)

, tmp_scan_all as (
  select
   network_type
  ,waybill_no
  ,scan_site
  ,scan_site_code
  ,dt
  ,row_number() over(partition by waybill_no,scan_site_code,network_type order by scan_time desc) rn
  from (
      select
       network_type
      ,waybill_no
      ,scan_site
      ,scan_site_code
      ,scan_time
      ,dt
      from tmp_scan_jisan  --集散点
      union all
      select
       '网点' as network_type
      ,waybill_no
      ,scan_site
      ,scan_site_code
      ,scan_time
      ,dt
      from jms_dwd.dwd_tab_barscan_deliver_base_dt t1   --网点派件表
      where dt >= date_add('{{ execution_date | cst_ds }}',-7) and dt <= '{{ execution_date | cst_ds }}'
      and waybill_no is not null
  ) t
)

--退回件表
, tmp_rolback as (
select
   waybill_no
  ,apply_type_code
  ,apply_type_name
  ,apply_time
  ,status
  ,status_name
from jms_dwd.dwd_tab_reback_transfer_express_base
where apply_type_code = '4' --退回揽收网点
  and waybill_no is not null
)

,tmp_2 as (
select
   dt
  ,scan_site_code
  ,nvl(max(scan_site),'空网点') as scan_site
  ,count(1) as bill_cnt
  ,sum(is_rolback_flag) as bill_rollback_cnt
from (
  select
     t1.network_type
    ,t1.waybill_no
    ,t1.scan_site
    ,t1.scan_site_code
    ,if(t2.waybill_no is not null,1,0) is_rolback_flag  --退件标识
    ,dt
  from tmp_scan_all t1
  left join tmp_rolback t2
         on t1.waybill_no = t2.waybill_no
      where rn=1
) t
group by scan_site_code,dt
)

select
 dt
,scan_site
,bill_cnt
,bill_rollback_cnt
,bill_rollback_cnt/bill_cnt as bill_rollback_rate
from tmp_2
;
 " >>/tmp/${name}_${dt}.txt
}

# 替换分隔符，并输出csv
cat /tmp/${name}_${dt}.txt | tr -s '\t' ',' > /tmp/${name}_${dt}.csv
# 压缩并删除 csv 文件
zip -jrm /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv
# 上传文件
/usr/local/python3/bin/python3  /usr/local/airflow/dags/utils/operators/upload_oss.py ${name}/${name}_${dt}.zip /tmp/${name}_${dt}.zip
# zip 文件将在进程退出后被自动清理

#声明文件大小
data_size=$(ls -lh /tmp/${name}_${dt}.zip | awk '{print $5}')
#打印文件信息
Message="${name}_${dt}.zip 上传成功 大小: $data_size"
#配置钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=816a8f35fd84432ffa6bff9d7714abc322628bb4a8ded86908550d2335aba613' \
   -H 'Content-Type: application/json' \
   -d "
  {\"msgtype\": \"text\",
    \"text\": {
        \"content\": \"监控内容:$Message\"
     }
  }" || true
rm -f  /tmp/${name}_${dt}.txt